package com.wpp.mq.kafka.controller;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

/**
 * @description:
 * @author: WangPengPeng
 * @create: 2021-12-12 20:55
 **/
@RestController
@RequestMapping("/api/kafka/sync")
public class KafkaSyncProducerController {

    @Resource
    private KafkaTemplate<Integer, String> template;

    @RequestMapping("send/{message}/{topic}")
    public String send(@PathVariable String message,@PathVariable String topic) {

        final ListenableFuture<SendResult<Integer, String>> future = template.send(StringUtils.isEmpty(topic) ?"topic-spring-01" : topic, 0, 0, message);
        // 同步发送消息
        try {
            final SendResult<Integer, String> sendResult = future.get();
            final RecordMetadata metadata = sendResult.getRecordMetadata();

            System.out.println("topic : "+metadata.topic() + "\t partition:  " + metadata.partition() + "\t offset: " + metadata.offset());

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        return "success";
    }

}